package com.atguigu.day07;

import com.atguigu.bean.Event;
import com.atguigu.day03.Flink05_Source_Custom;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.HashMap;

public class Flink01_ProcessAllWindowTopN {
    public static void main(String[] args) throws Exception {
        //1.获取流的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.获取数据
        DataStreamSource<Event> streamSource = env.addSource(new Flink05_Source_Custom.ClickSource());

        //3.指定事件时间&WaterMark
        SingleOutputStreamOperator<Event> singleOutputStreamOperator = streamSource.assignTimestampsAndWatermarks(WatermarkStrategy.<Event>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event element, long recordTimestamp) {
                        return element.timestamp;
                    }
                })
        );

        //4.提取url
        SingleOutputStreamOperator<String> urlDStream = singleOutputStreamOperator.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event value) throws Exception {
                return value.url;
            }
        });

        //5.做keyby,将所有的url数据放到同一个分区
        KeyedStream<String, String> keyedStream = urlDStream.keyBy(new KeySelector<String, String>() {
            @Override
            public String getKey(String value) throws Exception {
                return "key";
            }
        });

        //需要统计最近10秒钟内最热门的两个url链接，并且每5秒钟更新一次

        //6.开启一个基于事件的滑动窗口，窗口大小为10s，滑动步长为5s
        WindowedStream<String, String, TimeWindow> window = keyedStream.window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)));

        //7.求窗口中最热门的两个url
        SingleOutputStreamOperator<Tuple3<String, Integer, Integer>> result = window.process(new MyTopNProcess(10));

        result.print();


        env.execute();

    }

    public static class MyTopNProcess extends ProcessWindowFunction<String, Tuple3<String,Integer,Integer>, String, TimeWindow>{

        //控制循环的条件也就是Topn中的n
        private Integer loopCtr;

        public MyTopNProcess(Integer loopCtr) {
            this.loopCtr = loopCtr;
        }

        @Override
        public void process(String s, ProcessWindowFunction<String, Tuple3<String, Integer, Integer>, String, TimeWindow>.Context context, Iterable<String> elements, Collector<Tuple3<String, Integer, Integer>> out) throws Exception {
            HashMap<String, Integer> urlCount = new HashMap<>();

            //1.将这个窗口中相同的url进行计数
            for (String element : elements) {
                if (urlCount.containsKey(element)){
                    //取出之前已经计算过的个数然后进行更新
                    Integer lastCount = urlCount.get(element);
                    lastCount = lastCount + 1;
                    urlCount.put(element, lastCount);
                }else {
                    //证明这个url是第一次进来
                    urlCount.put(element, 1);
                }
            }

            //2.针对不同url的个数进行排序，由大到小排序
            //2.1利用List集合进行排序，那么就意味着要将HashMap中的数据遍历出来放入List集合
            ArrayList<Tuple2<String, Integer>> sortList = new ArrayList<>();
            for (String url : urlCount.keySet()) {
                sortList.add(Tuple2.of(url, urlCount.get(url)));
            }

            //2.2对List集合中的数据做排序，由大到小
            sortList.sort(new Comparator<Tuple2<String, Integer>>() {
                @Override
                public int compare(Tuple2<String, Integer> o1, Tuple2<String, Integer> o2) {
                    return o2.f1-o1.f1;
                }
            });

            if (loopCtr<=sortList.size()){
                //3.取出这个窗口中排名最高的两个url
                for (int i = 0; i < loopCtr; i++) {
                    out.collect(Tuple3.of(sortList.get(i).f0,sortList.get(i).f1,i+1));
                }
            }else {
                System.out.println("请注意！！！！  给的参数不对");
            }
        }
    }
}
